package com.jy.rhin.application.basic;

import com.jy.framework.base.entity.Result;
import com.jy.rhin.application.assembler.DocumentSubAndPubAssembler;
import com.jy.rhin.application.infra.NotificationApplicationService;
import com.jy.rhin.infra.exception.NotifyException;
import com.jy.rhin.sdk.dto.notification.req.SavePublishMessageReqDto;
import com.jy.rhin.sdk.dto.notification.resp.PublishMessageDto;
import com.jy.rhin.support.notification.NotificationCallFactory;
import com.jy.rhin.support.notification.client.NotificationClient;
import lombok.extern.slf4j.Slf4j;
import org.oasis_open.docs.wsn.b_2.Subscribe;
import org.oasis_open.docs.wsn.b_2.SubscribeResponse;
import org.oasis_open.docs.wsn.b_2.Unsubscribe;
import org.oasis_open.docs.wsn.b_2.UnsubscribeResponse;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author WeiQ
 * @date 2022/7/6 10:00
 */
@Slf4j
@Service
public class NotificationService {

    @Resource
    NotificationApplicationService notificationApplicationService;
    @Resource
    RedissonClient redissonClient;

    @Resource
    NotificationCallFactory notificationCallFactory;


    /**
     * lock锁key
     */
    private static final String NOTIFY_SUBSCRIBER_LOCK = "com.jy.rhin.aggregation.notify.subscriber.opt.lock:";

    /**
     * lock锁自动释放时间
     */
    private static final int LEASE_TIME = 10;

    /**
     * 通知订阅者，通知接口调用失败需要重试，最多重试三次
     */
    @Retryable(maxAttempts = 4)
    @Async("publishMessageRetryExecutor")
    public void notifySubscriber(PublishMessageDto publishMessage) {
        Long messageLogId = publishMessage.getId();
        RLock lock = redissonClient.getLock(NOTIFY_SUBSCRIBER_LOCK + messageLogId);
        try {
            log.debug("给订阅者发送通知获取锁");
            if (lock.tryLock(0, LEASE_TIME, TimeUnit.SECONDS)) {
                log.debug("开始给订阅者发送通知");
                //获取对应消息通知调用客户端实现类，通知订阅者
                NotificationClient client = notificationCallFactory.getClient(publishMessage.getInteractionType());
                client.notify(publishMessage);
                //通知成功，调用基础服务接口保存通知结果
                saveSuccessPublishMessage(messageLogId);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new NotifyException(messageLogId, e);
        } catch (Exception e) {
            throw new NotifyException(messageLogId, e);
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }

    }


    /**
     * 异常重试，超过重试后调用此方法
     *
     * @param exception
     */
    @Recover
    @SuppressWarnings("unused")
    public void notifySubscriberFallback(NotifyException exception) {
        //通知订阅者失败，调用基础服务接口保存通知结果
        log.error("通知订阅者失败，开始保存通知结果", exception);
        saveFailPublishMessage(exception);
    }



    /**
     * 订阅
     * @return
     */
    public SubscribeResponse subscribe(Subscribe subscribeRequest) {
        Result<Long> result =Result.success( notificationApplicationService.businessSubscribeTopic(DocumentSubAndPubAssembler.toDto(subscribeRequest)));

        SubscribeResponse resp = new SubscribeResponse();
        if (result.isSuccess()) {
            resp.setSuccessMessge(result.getMsg());
        } else {
            resp.setFault(result.getMsg());
        }
        return resp;
    }

    /**
     * 取消订阅
     * @return
     */
    public UnsubscribeResponse unsubscribe(Unsubscribe unsubscribeRequest) {
        Result<Boolean> result = Result.success(notificationApplicationService.cancelSubscribe(DocumentSubAndPubAssembler.toDto(unsubscribeRequest)));
        UnsubscribeResponse resp = new UnsubscribeResponse();
        if (result.isSuccess()) {
            resp.setSuccessMessge(result.getMsg());
        } else {
            resp.setFault(result.getMsg());
        }
        return resp;
    }


    /**
     * 调用成功通知调用结果
     *
     * @param messageId 消息记录id
     * @return
     */
    private void saveSuccessPublishMessage(Long messageId) {
        SavePublishMessageReqDto req = new SavePublishMessageReqDto();
        req.setPublishId(messageId);
        req.setPublishResult(true);
        notificationApplicationService.publishMessage(req.getPublishId(), req.getPublishResult(), req.getFailReason());

    }

    /**
     * 调用失败通知调用结果
     *
     * @param exception
     * @return
     */
    private void saveFailPublishMessage(NotifyException exception) {
        SavePublishMessageReqDto req = new SavePublishMessageReqDto();
        req.setPublishId(exception.getMessageId());
        req.setPublishResult(false);
        req.setFailReason(exception.getMessage());
        notificationApplicationService.publishMessage(req.getPublishId(), req.getPublishResult(), req.getFailReason());

    }
}
